package net.xo68.destiny.core.timingwheel;

import lombok.extern.slf4j.Slf4j;
import net.xo68.destiny.core.timingwheel.impl.TimingWheelTimeout;

import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * 时光轮桶(线程安全阻塞队列)
 * ConcurrentLinkedQueue 单线程生产，多线程消费
 * LinkedBlockingQueue  多线程生产，单线程消费
 * @author wuxie
 * @version 2020-10-18
 */
@Slf4j
public class TimingWheelBucket<T> extends ConcurrentLinkedQueue<Timeout<T>> {

    /**
     * 时光轮格子对应的桶索引
     */
    private final long bucketIndex;

    public TimingWheelBucket(long bucketIndex) {
        super();
        this.bucketIndex=bucketIndex;
    }

    public long bucketIndex(){
        return this.bucketIndex;
    }

    @Override
    public boolean offer(Timeout<T> timingWheelTimeout) {
        timingWheelTimeout.bucket(this);
        return super.offer(timingWheelTimeout);
    }

    @Override
    public boolean remove(Object o) {
        if(o instanceof TimingWheelTimeout){
            ((Timeout) o).bucket(null);
        }
        return super.remove(o);
    }

    /**
     * 检查过期，处理超时项轮数，触发已过期的超时项，同时移除过期或取消项
     * Expire all {@link TimingWheelTimeout}s for the given {@code deadline}.
     * @param deadline 延迟时间
     */
    public void expire(long deadline) {
        Iterator<Timeout<T>> iterator = iterator();
        while (iterator.hasNext()){
            Timeout<T> timeout=iterator.next();
            if (timeout.remainingRounds() <= 0) {
                //从桶中移除
                iterator.remove();
                timeout.bucket(null);
                if (timeout.deadline() <= deadline) {
                    timeout.expire();
                    //执行完后，从缓冲移除
                    //timeout.remove();
                } else {
                    // The timeout was placed into a wrong slot. This should never happen.
                    throw new IllegalStateException(String.format("timeout.deadline (%d) > deadline (%d)",
                            timeout.deadline(), deadline));
                }
            } else if (timeout.isCancelled()) {
                iterator.remove();
                timeout.bucket(null);
                timeout.remove();
            } else {
                timeout.remainingRoundDecrement();
            }
        }
        //log.debug(" > >  > > bucketIndex: {},size: {}",this.bucketIndex,this.size());
    }



    /**
     * Clear this bucket and return all not expired / cancelled {@link TimingWheelTimeout}s.
     */
    public long clearTimeouts(Set<Timeout<T>> set) {
        long removeCount=0L;
        Iterator<Timeout<T>> iterator = iterator();
        while (iterator.hasNext()){
            Timeout<T> next=iterator.next();
            iterator.remove();
            removeCount++;
            next.bucket(null);
            if (next.isExpired() || next.isCancelled()) {
                continue;
            }
            set.add(next);
        }
        return removeCount;
    }

}
